package cn.cnooc.config;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDB.ConsistencyLevel;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.*;
import org.influxdb.dto.Point.Builder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * InfluxDB数据库连接操作类
 */
@Data
@Slf4j
@Component()
public class InfluxDbComUtil {
    // 用户名
    @Value("${influxdb.username}")
    private String username;
    // 密码
    @Value("${influxdb.password}")
    private String password;
    // 连接地址
    @Value("${influxdb.url}")
    private String openurl;
    // 数据库名称
    @Value("${influxdb.database}")
    private String dbName;
    // 保留策略
    @Value("${influxdb.retention}")
    private String retentionPolicy;

    private InfluxDB influxDB;

    public InfluxDbComUtil() {
    }

    /**
     * 连接时序数据库 ，若不存在则创建
     *
     * @return
     */
    @Bean
    public InfluxDB influxDbBuild() {
        if (influxDB == null) {
            influxDB = InfluxDBFactory.connect(openurl, username, password);
        }
        try {
            if (!influxDB.databaseExists(dbName)) {
                influxDB.createDatabase(dbName);
            }
        } catch (Exception e) {
            // 该数据库可能设置动态代理，不支持创建数据库
            e.printStackTrace();
        } finally {
            influxDB.setRetentionPolicy(retentionPolicy);
        }
        influxDB.setLogLevel(InfluxDB.LogLevel.NONE);
        return influxDB;
    }

    /**
     * 测试连接是否正常
     *
     * @return true 正常
     */
    public boolean ping() {
        boolean isConnected = false;
        Pong pong;
        try {
            pong = influxDB.ping();
            if (pong != null) {
                isConnected = true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return isConnected;
    }

    /**
     * 创建数据库
     *
     * @param dbName
     */
    @SuppressWarnings("deprecation")
    public void createDB(String dbName) {
        influxDB.createDatabase(dbName);
    }

    /**
     * 删除数据库
     *
     * @param dbName
     */
    @SuppressWarnings("deprecation")
    public void deleteDB(String dbName) {
        influxDB.deleteDatabase(dbName);
    }

    /**
     * 切换数据库
     */
    @SuppressWarnings("deprecation")
    public void setDB(String dbName) {
        influxDB.setDatabase(dbName);
    }

    /**
     * 关闭数据库
     */
    public void close() {
        influxDB.close();
    }

    /**
     * 切换数据库策略
     *
     * @param dataBaseName 数据库
     * @param policyName   策略名
     */
    public void updRetentionPolicy(String dataBaseName, String policyName) {
        String sql = String.format("ALTER RETENTION POLICY \"" + policyName + "\" ON \"" + dataBaseName + "\" DEFAULT");
        query(sql);
        this.dbName = dataBaseName;
        this.retentionPolicy = policyName;
    }

    /**
     * 切换策略
     *
     * @param policyName 策略名
     */
    public void updRetentionPolicy(String policyName) {
        String sql = String.format("ALTER RETENTION POLICY \"" + policyName + "\" ON \"" + dbName + "\" DEFAULT");
        query(sql);
        this.retentionPolicy = policyName;
    }

    /**
     * 创建自定义保留策略
     *
     * @param policyName  策略名
     * @param days        保存天数
     * @param replication 保存副本数量
     * @param isDefault   是否设为默认保留策略
     */
    public void createRetentionPolicy(String dataBaseName, String policyName, int days, int replication, Boolean isDefault) {
        String sql = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %sd REPLICATION %s ", policyName,
                dataBaseName, days, replication);
        if (isDefault) {
            sql = sql + " DEFAULT";
        }
        query(sql);
    }

    /**
     * 创建默认的保留策略
     * <p>
     * 策略名：hour，保存天数：30天，保存副本数量：1,设为默认保留策略
     */
    public void createDefaultRetentionPolicy() {
        String command = String
                .format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT", "hour", dbName,
                        "30d", 1);
        this.query(command);
    }

/*********************************增删查**************************************************/
    /**
     * 查询
     *
     * @param command 查询语句
     * @return
     */
    public List<Map<String, Object>> query(String command) {
        log.info(command);
        return resultConversion(influxDB.query(new Query(command, dbName)));
    }


    /**
     * 插入
     *
     * @param measurement 表
     * @param tags        标签
     * @param fields      字段
     */
    public void insert(String measurement, Map fields, Map tags) {
        Builder builder = Point.measurement(measurement);
        builder.tag(tags);
        builder.fields(fields);
        influxDB.write(dbName, retentionPolicy, builder.build());
    }

    /**
     * 删除
     *
     * @param command 删除语句
     * @return 返回错误信息
     */
    public String deleteMeasurementData(String command) {
        QueryResult result = influxDB.query(new Query(command, dbName));
        return result.getError();
    }

    /**
     * 构建Point
     *
     * @param measurement 表
     * @param time        时间
     * @param timeUnit    延迟
     * @param tags        tags
     * @param fields
     * @return
     */
    public Point pointBuilder(String measurement, long time, TimeUnit timeUnit, Map<String, String> tags, Map<String, Object> fields) {
        Point point = Point.measurement(measurement).time(time, timeUnit).tag(tags).fields(fields).build();
        return point;
    }

    /**
     * 批量写入测点
     *
     * @param batchPoints
     */
    public void batchInsert(BatchPoints batchPoints, TimeUnit timeUnit) {
        influxDB.write(batchPoints);
        // influxDB.enableGzip();
        // influxDB.enableBatch(2000,100,timeUnit);
        // influxDB.disableGzip();
        // influxDB.disableBatch();
    }

    /**
     * 批量写入数据
     *
     * @param database        数据库
     * @param retentionPolicy 保存策略
     * @param consistency     一致性
     * @param records         要保存的数据（调用BatchPoints.lineProtocol()可得到一条record）
     */
    public void batchInsert(final String database, final String retentionPolicy, final ConsistencyLevel consistency, TimeUnit timeUnit, final List<String> records) {
        influxDB.write(database, retentionPolicy, consistency, timeUnit, records);
    }

    private List<Map<String, Object>> resultConversion(QueryResult queryResult) {
        List<Map<String, Object>> resultList = new ArrayList<>();
        queryResult.getResults().forEach(result -> {
                    if (result.getSeries()!=null) {
                        result.getSeries().forEach(series -> {
                            List<String> columns = series.getColumns();
                            List<List<Object>> values = series.getValues();
                            values.forEach(value -> {
                                        Map<String, Object> a = new LinkedHashMap<>();
                                        IntStream.range(0, value.size()).forEach(i ->
                                                a.put(columns.get(i), value.get(i))
                                        );
                                        resultList.add(a);
                                    }
                            );
                        });
                    }
                }

        );
        return resultList;
    }
}